package com.deep.flink.pgsql;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @author sw
 * @create 2023-06-30 11:26
 */
public class Test11 {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);


        tableEnv.executeSql("CREATE TABLE flink_cdc_source (\n" +
                "  id INT,\n" +
                "  name STRING\n" +
                ") WITH (\n" +
                "  'connector' = 'postgres-cdc',\n" +
                "  'hostname' = 'localhost',\n" +
                "  'port' = '5432',\n" +
                "  'database-name' = 'postgres',\n" +
                "  'schema-name' = 'public',\n" +
                "  'username' = 'postgres',\n" +
                "  'password' = '123456',\n" +
                "  'table-name' = 'test1',\n" +
                "  'decoding.plugin.name' = 'pgoutput'\n" +
                ");\n ");


        tableEnv.executeSql("CREATE TABLE flink_cdc_sink (\n" +
                "  id INT,\n" +
                "  name STRING,\n" +
                "  PRIMARY KEY (id) NOT ENFORCED\n" +
                ") WITH (\n" +
                "  'connector' = 'jdbc',\n" +
                "  'url' = 'jdbc:postgresql://localhost:5432/postgres',\n" +
                "  'username' = 'postgres',\n" +
                "  'password' = '123456',\n" +
                "  'table-name' = 'test2'\n" +
                ");\n");


        tableEnv.executeSql("insert into flink_cdc_sink select * from flink_cdc_source");

        env.execute();

    }
}
